package cn.getech.data.development.utils

import java.io.{File, FileOutputStream, InputStream, OutputStreamWriter, PrintWriter}

import cn.getech.data.development.bean.jdbc.JDBCFieldBean
import com.alibaba.fastjson.{JSONArray, JSONObject}
import org.apache.avro.Schema
import org.slf4j.{Logger, LoggerFactory}

import scala.io.Source


object AVSCFileUtils {

  private val logger: Logger = LoggerFactory.getLogger(this.getClass)

  /**
   * 生成avsc文件
   *
   * @param fields 生成avsc的参数
   * @param fileName 生成avsc的文件名
   * @return
   */
  def apply(fields: Array[JDBCFieldBean], fileName: String): Boolean = {
    val obj = new JSONObject()
    val subObjs = new JSONArray()
    obj.put("namespace", "cn.getech.data.development.bean.hdfs.parquet")
    obj.put("type", "record")
    obj.put("name", "HDFSParquetFileDynamicBean")
    val utils = new JDBCSQLTypeUtils
    fields.foreach(x => {
      val nObject = new JSONObject()
      nObject.put("name", x.field_name)
      nObject.put("type", utils.parquet_type2String(x.class_type))
      subObjs.add(nObject)
    })
    obj.put("fields", subObjs)
    val flag = createFile(obj.toJSONString, fileName)
    logger.info("The parquet file was created successfully: ", flag)
    flag
  }

  /**
   * 获取文件生成schema
   * @param fileName
   * @return schema
   */
  def schema(fileName: String) = new Schema.Parser().parse(getClass.getClassLoader.getResourceAsStream(fileName))

  /**
   * 根据objs生成schema
   * @param fields
   * @return schema
   */
  def schema(fields: Array[JDBCFieldBean]) = {
    val obj = new JSONObject()
    val subObjs = new JSONArray()
    obj.put("namespace", "cn.getech.data.development.bean.hdfs.parquet")
    obj.put("type", "record")
    obj.put("name", "HDFSParquetFileDynamicBean")
    val utils = new JDBCSQLTypeUtils
    fields.foreach(x => {
      val nObject = new JSONObject()
      nObject.put("name", x.field_name)
      nObject.put("type", utils.parquet_type2String(x.class_type))
      subObjs.add(nObject)
    })
    obj.put("fields", subObjs)
    new Schema.Parser().parse(obj.toJSONString)
  }

  def readerStream2String(input: InputStream): String = Source.fromInputStream(input).mkString

  def printFile(text: String, filePath: String): Boolean = {
    var flag = true
    var writer: PrintWriter = null
    try {
      writer = new PrintWriter(new File(filePath))
      writer.println(text)
    } catch {
      case e: Exception =>
        flag = false
        e.printStackTrace()
    } finally {
      if (writer != null)
        writer.close()
    }
    flag
  }

  /**
   * 创建文件
   *
   * @param text     输入文本
   * @param filePath 文件路径
   * @return flag
   */
  def createFile(text: String, filePath: String): Boolean = {
    // 标记文件生成是否成功
    var flag = true
    var write: OutputStreamWriter = null
    try {
      val file = new File(filePath)
      if (!file.getParentFile.exists) {
        file.getParentFile.mkdirs
      }
      if (file.exists) {
        file.delete // 删除旧文件
      }
      file.createNewFile
      // 将格式化后的字符串写入文件
      write = new OutputStreamWriter(new FileOutputStream(file), "UTF-8")
      write.write(text)
      write.flush()
    } catch {
      case e: Exception =>
        flag = false
        if (write != null)
          write.close()
        e.printStackTrace()
    } finally {
      if (write != null)
        write.close()
    }
    flag
  }
}
